package org.luxor.accesslog.service.impl;


import org.elasticsearch.ElasticsearchException;
import org.elasticsearch.action.ActionListener;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.index.IndexResponse;
import org.elasticsearch.action.search.SearchRequest;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.common.xcontent.XContentBuilder;
import org.elasticsearch.common.xcontent.XContentFactory;
import org.elasticsearch.common.xcontent.XContentType;
import org.elasticsearch.index.query.BoolQueryBuilder;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.index.reindex.BulkByScrollResponse;
import org.elasticsearch.index.reindex.DeleteByQueryRequest;
import org.elasticsearch.rest.RestStatus;
import org.elasticsearch.search.SearchHit;
import org.elasticsearch.search.builder.SearchSourceBuilder;
import org.elasticsearch.search.sort.SortOrder;
import org.luxor.accesslog.config.AccessLogProperties;
import org.luxor.accesslog.entity.AccessLogEntity;
import org.luxor.accesslog.entity.AccessLogQuery;
import org.luxor.accesslog.service.IAccessLogService;
import org.luxor.commons.core.utils.DateUtils;
import org.luxor.commons.core.utils.Jackson2Utils;
import org.luxor.commons.core.utils.StringUtils;
import org.luxor.commons.core.utils.UuidUtils;
import org.luxor.commons.core.validator.ValidatorUtils;
import org.luxor.commons.core.validator.group.AddGroup;
import org.luxor.commons.elasticsearch.ElasticClientRest;
import org.luxor.commons.elasticsearch.metadata.EsOrderItem;
import org.luxor.commons.elasticsearch.metadata.EsPageQuery;
import org.luxor.commons.elasticsearch.metadata.EsPageResult;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import javax.annotation.PostConstruct;
import java.io.IOException;
import java.util.Arrays;
import java.util.Date;
import java.util.List;
import java.util.stream.Collectors;

/**
 * OSS配置服务
 *
 * @author Mr.yan @date 2020/3/14
 */
@Service("accessLogService")
public class AccessLogServiceImpl implements IAccessLogService {

    private static final Logger logger = LoggerFactory.getLogger(AccessLogServiceImpl.class);

    @Autowired
    private ElasticClientRest elasticClientRest;

    @Autowired
    private AccessLogProperties properties;

    @PostConstruct
    public void initIndex() throws ElasticsearchException {
        try {
            XContentBuilder builder = XContentFactory.jsonBuilder();
            builder.startObject();
            {
                builder.startObject("properties");
                {
                    builder.startObject("id");
                    {
                        builder.field("type", "keyword");
                    }
                    builder.endObject();
                    builder.startObject("userName");
                    {
                        builder.field("type", "keyword");
                        builder.field("ignore_above", 50);
                    }
                    builder.endObject();
                    builder.startObject("ip");
                    {
                        builder.field("type", "ip");
                    }
                    builder.endObject();
                    builder.startObject("module");
                    {
                        builder.field("type", "text");
                    }
                    builder.endObject();
                    builder.startObject("message");
                    {
                        builder.field("type", "text");
                    }
                    builder.endObject();
                    builder.startObject("time");
                    {
                        builder.field("type", "date");
                        builder.field("format", DateUtils.DATE_TIME_PATTERN);
                    }
                    builder.endObject();
                }
                builder.endObject();
            }
            builder.endObject();
            elasticClientRest.createIndex(properties.getIndexName(), builder);
        } catch (IOException e) {
            throw new ElasticsearchException("创建 [{}] 索引时,出现异常:{}", properties.getIndexName(), e);
        }
    }

    @Override
    public void saveLog(AccessLogEntity accessLog) {
        if (StringUtils.isBlank(accessLog.getId())) {
            accessLog.setId(UuidUtils.uuid32());
        }
        // 校验参数合法性
        ValidatorUtils.validateEntity(accessLog, AddGroup.class);

        IndexRequest request = new IndexRequest(properties.getIndexName());
        request.id(accessLog.getId());
        request.source(Jackson2Utils.toJson(accessLog), XContentType.JSON);
        elasticClientRest.rest.indexAsync(request, RequestOptions.DEFAULT, new ActionListener<IndexResponse>() {
            @Override
            public void onResponse(IndexResponse updateResponse) {
                logger.debug("write [{}] into elastic succ", updateResponse.getIndex());
            }

            @Override
            public void onFailure(Exception e) {
                logger.debug("write to elastic error log:{}", accessLog);
                logger.error("write to elastic error", e);
            }
        });
    }

    @Override
    public EsPageResult<AccessLogEntity> pageContent(EsPageQuery<AccessLogQuery> query) {
        // 分页条件
        SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();
        searchSourceBuilder.from((query.getCurrent() - 1) * query.getSize());
        searchSourceBuilder.size(query.getSize());

        // 获取过滤条件
        AccessLogQuery condition = query.getSearch();
        if (condition != null) {
            // 搜索过滤条件
            BoolQueryBuilder boolQueryBuilder = QueryBuilders.boolQuery();
            if (condition.getBeginTime() != null && condition.getBeginTime().getTime() > 0) {
                boolQueryBuilder.must(QueryBuilders.rangeQuery("time").from(DateUtils.format(condition.getBeginTime())));
            }
            if (condition.getEndTime() != null && condition.getEndTime().getTime() > 0) {
                boolQueryBuilder.must(QueryBuilders.rangeQuery("time").to(DateUtils.format(condition.getEndTime())));
            }

            if (StringUtils.isNotBlank(condition.getUserName())) {
                boolQueryBuilder.must(QueryBuilders.matchQuery("userName", condition.getUserName()));
            }
            if (StringUtils.isNotBlank(condition.getMessage())) {
                boolQueryBuilder.must(QueryBuilders.wildcardQuery("message", condition.getMessage()));
            }
            searchSourceBuilder.query(boolQueryBuilder);
        }

        // 搜索排序条件
        EsOrderItem orderCondition = query.getOrders();
        if (orderCondition != null) {
            SortOrder order = orderCondition.getAsc() ? SortOrder.ASC : SortOrder.DESC;
            searchSourceBuilder.sort(orderCondition.getColumn(), order);
        }

        // 执行分页查询
        EsPageResult<AccessLogEntity> pageData = new EsPageResult<>();
        try {
            SearchRequest searchRequest = new SearchRequest(properties.getIndexName());
            searchRequest.source(searchSourceBuilder);
            SearchResponse searchResponse = elasticClientRest.rest.search(searchRequest, RequestOptions.DEFAULT);
            if (RestStatus.OK.equals(searchResponse.status())) {
                // 总记录数
                long total = searchResponse.getHits().getTotalHits().value;
                // 分页数据
                SearchHit[] searchHits = searchResponse.getHits().getHits();
                // 组装分页对象
                pageData.setScrollId(searchResponse.getScrollId());
                pageData.setCurrent(query.getCurrent());
                pageData.setSize(searchHits.length);
                pageData.setTotal(total);
                pageData.setTotalPage(total / query.getSize());
                List<AccessLogEntity> records = Arrays.stream(searchHits).map(hit ->
                        Jackson2Utils.toBean(hit.getSourceAsString(), AccessLogEntity.class)
                ).collect(Collectors.toList());
                pageData.setRecords(records);
            }
            return pageData;
        } catch (IOException e) {
            throw new ElasticsearchException("分页查询 [{}] 索引中的文档内容时,出现异常:{}", properties.getIndexName(), e);
        }
    }

    @Override
    public void flush() {
        DeleteByQueryRequest deleteRequest = new DeleteByQueryRequest(properties.getIndexName());
        deleteRequest.setConflicts("proceed");
        Date maxTime = new Date(System.currentTimeMillis() - properties.getTtlSec() * 1000L);
        deleteRequest.setQuery(QueryBuilders.boolQuery().filter(QueryBuilders.rangeQuery("time").lte(DateUtils.format(maxTime))));

        elasticClientRest.rest.deleteByQueryAsync(deleteRequest, RequestOptions.DEFAULT, new ActionListener<BulkByScrollResponse>() {

            @Override
            public void onResponse(BulkByScrollResponse bulkByScrollResponse) {
                logger.debug("delete Doc from elastic.index [{}] succ.", properties.getIndexName());
            }

            @Override
            public void onFailure(Exception e) {
                logger.debug("delete Doc from elastic.index [{}] error log:{}", properties.getIndexName(), e);
                logger.error("delete Doc from elastic.index [" + properties.getIndexName() + "] error", e);
            }
        });
    }
}
